//package com.cloud2022.order.canal;
//
//import com.alibaba.otter.canal.client.CanalConnector;
//import com.alibaba.otter.canal.client.CanalConnectors;
//import com.alibaba.otter.canal.protocol.CanalEntry;
//import com.alibaba.otter.canal.protocol.Message;
//import com.google.protobuf.InvalidProtocolBufferException;
//import org.springframework.beans.factory.InitializingBean;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.config.BeanPostProcessor;
//import org.springframework.scheduling.annotation.Async;
//import org.springframework.scheduling.annotation.Scheduled;
//import org.springframework.stereotype.Component;
//
//import java.io.File;
//import java.io.FileOutputStream;
//import java.net.InetSocketAddress;
//import java.util.ArrayList;
//import java.util.Collection;
//import java.util.List;
//
///**
// * @author: create by God of SJF
// * @version: v1.0
// * @description: com.cloud2022.order.canal
// * @date:2022/2/11
// */
//
//@Component
//public class CanalListener {
//
//    @Autowired
//    CanalConnectorBean canalConnectorBean;
//
//    @Scheduled(cron = "0/1 * * * * ? ")
//    public void getChange() throws Exception {
//        File file = new File("/Users/libin/Desktop/小糊涂仙营销Binlog");
//        // 创建链接，connector也是canal数据操作客户端
//        CanalConnector connector = canalConnectorBean.connector;
//        int batchSize = 1000;
//        int emptyCount = 0;
//        try {
//            // 链接对应的canal server
//            connector.connect();
//            // 客户端订阅，重复订阅时会更新对应的filter信息，这里订阅所有库的所有表
//            connector.subscribe("record._record");
//            // 回滚到未进行 ack 的地方，下次fetch的时候，可以从最后一个没有 ack 的地方开始拿
//            connector.rollback();
//            int totalEmptyCount = 120;
//            // 循环遍历120次
//            // 尝试拿batchSize条记录，有多少取多少，不会阻塞等待
//            Message message = connector.getWithoutAck(batchSize);
//            // 消息id
//            long batchId = message.getId();
//            // 实际获取记录数
//            int size = message.getEntries().size();
//            // 如果没有获取到消息
//            if (batchId == -1 || size == 0) {
//                emptyCount++;
//                System.out.println("empty count : " + emptyCount);
//                try {
//                    Thread.sleep(1000);
//                } catch (InterruptedException e) {
//                }
//            } else {
//                // 如果消息不为空，重置遍历。从0开始重新遍历
//                emptyCount = 0;
//                // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
//                printEntry(message.getEntries(), file);
//            }
//
//            // 进行 batch id 的确认。
//            connector.ack(batchId); // 提交确认
//            // 回滚到未进行 ack 的地方，指定回滚具体的batchId；如果不指定batchId，回滚到未进行ack的地方
//            // connector.rollback(batchId); // 处理失败, 回滚数据
//
////            System.out.println("empty too many times, exit");
//        } finally {
//            // 释放链接
//            connector.disconnect();
//        }
//    }
//
//    private static void printEntry(List<CanalEntry.Entry> entrys, File file) {
//        for (CanalEntry.Entry entry : entrys) {
//            // 如果是事务操作，直接忽略。 EntryType常见取值：事务头BEGIN/事务尾END/数据ROWDATA
//            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
//                continue;
//            }
//            CanalEntry.RowChange rowChange = null;
//            try {
//                // 获取byte数据，并反序列化
//                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
//            } catch (Exception e) {
//                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
//                        e);
//            }
//
//            CanalEntry.EventType eventType = rowChange.getEventType();
//
//            System.out.println("====================================begin========================================");
//            System.out.println(String.format("基本信息 binlog[%s:%s] , 表[%s.%s] , 操作: %s",
//                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
//                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
//                    eventType));
//
//
//            // 如果是ddl或者是查询操作，直接打印sql
//            System.out.println(rowChange.getSql() + ";");
//
//            // 如果是删除、更新、新增操作解析出数据
////            for (RowData rowData : rowChange.getRowDatasList()) {
//            if (eventType == CanalEntry.EventType.DELETE) {
//                // 删除操作，只有删除前的数据
//                saveDeleteSql(entry, file);
////                    printColumn(rowData.getBeforeColumnsList());
//            } else if (eventType == CanalEntry.EventType.INSERT) {
//                // 新增数据，只有新增后的数据
////                    printColumn(rowData.getAfterColumnsList());
//                saveInsertSql(entry, file);
//            } else {
////                    // 更新数据：获取更新前后内容
////                    System.out.println("-------> before");
////                    printColumn(rowData.getBeforeColumnsList());
////                    System.out.println("-------> after");
////                    printColumn(rowData.getAfterColumnsList());
//                saveUpdateSql(entry, file);
//            }
////            }
//            System.out.println("------------------------------------end------------------------------------------");
//        }
//    }
//
//    private static void printColumn(List<CanalEntry.Column> columns) {
//        File file = new File("/Users/libin/Desktop/小糊涂仙营销Binlog");
//        List<String> list = new ArrayList<>();
//        for (CanalEntry.Column column : columns) {
//            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
//            list.add(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated() + "\r\n");
//        }
//        try {
//            writeFile(file.getAbsolutePath(), list);
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//
//    public synchronized static void writeFile(String filePath, Collection<String> lineList) throws Exception {
//        File copy = new File(filePath.substring(0, filePath.lastIndexOf("/")));
////        File copy = new File(filePath);
//        if (!copy.exists()) {
//            copy.mkdirs();
//        }
//        try {
//            FileOutputStream fout = new FileOutputStream(filePath, true);
//            for (String string : lineList) {
//                fout.write(string.getBytes("utf-8"));
//            }
//            fout.flush();
//            fout.close();
//        } catch (Exception e) {
//            throw e;
//        }
//    }
//
//
//    private static void saveUpdateSql(CanalEntry.Entry entry, File file) {
//        try {
//            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
//            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
//            List<String> list = new ArrayList<>();
//            for (CanalEntry.RowData rowData : rowDatasList) {
//                List<CanalEntry.Column> newColumnList = rowData.getAfterColumnsList();
//                StringBuffer sql = new StringBuffer("update " + entry.getHeader().getTableName() + " set ");
//                for (int i = 0; i < newColumnList.size(); i++) {
//                    CanalEntry.Column column = newColumnList.get(i);
//                    if (column.getUpdated()) {
//                        sql.append(" " + newColumnList.get(i).getName()
//                                + " = '" + newColumnList.get(i).getValue() + "'");
//                        if (i != newColumnList.size() - 1) {
//                            sql.append(",");
//                        }
//                    }
//
//                }
//                sql.append(" where ");
//                List<CanalEntry.Column> oldColumnList = rowData.getBeforeColumnsList();
//                for (CanalEntry.Column column : oldColumnList) {
//                    if (column.getIsKey()) {
//                        //暂时只支持单一主键
//                        sql.append(column.getName() + "=" + column.getValue());
//                        break;
//                    }
//                }
//                String s = sql.toString();
//                int i = s.lastIndexOf(",");
//                //这样就去掉可最后一个逗号
//                if (i != -1) {
//                    String before = s.substring(0, i);
//                    String after = s.substring(i + 1);
//                    list.add(before + after + "\r\n");
//                } else {
//                    list.add(s + "\r\n");
//                }
//            }
//            try {
//                writeFile(file.getAbsolutePath(), list);
//            } catch (Exception e) {
//                e.printStackTrace();
//            }
//        } catch (InvalidProtocolBufferException e) {
//            e.printStackTrace();
//        }
//    }
//
//    private static void saveDeleteSql(CanalEntry.Entry entry, File file) {
//        try {
//            List<String> list = new ArrayList<>();
//            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
//            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
//            for (CanalEntry.RowData rowData : rowDatasList) {
//                List<CanalEntry.Column> columnList = rowData.getBeforeColumnsList();
//                StringBuffer sql = new StringBuffer("delete from " + entry.getHeader().getTableName() + " where ");
//                for (CanalEntry.Column column : columnList) {
//                    if (column.getIsKey()) {
//                        //暂时只支持单一主键
//                        sql.append(column.getName() + "=" + column.getValue());
//                        break;
//                    }
//                }
//                list.add(sql.toString() + "\r\n");
//            }
//            try {
//                writeFile(file.getAbsolutePath(), list);
//            } catch (Exception e) {
//                e.printStackTrace();
//            }
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//
//    private static void saveInsertSql(CanalEntry.Entry entry, File file) {
//        try {
//            List<String> list = new ArrayList<>();
//            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
//            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
//            for (CanalEntry.RowData rowData : rowDatasList) {
//                List<CanalEntry.Column> columnList = rowData.getAfterColumnsList();
//                StringBuffer sql = new StringBuffer("insert into " + entry.getHeader().getTableName() + " (");
//                for (int i = 0; i < columnList.size(); i++) {
//                    sql.append(columnList.get(i).getName());
//                    if (i != columnList.size() - 1) {
//                        sql.append(",");
//                    }
//                }
//                sql.append(") VALUES (");
//                for (int i = 0; i < columnList.size(); i++) {
//                    sql.append("'" + columnList.get(i).getValue() + "'");
//                    if (i != columnList.size() - 1) {
//                        sql.append(",");
//                    }
//                }
//                sql.append(")");
//                list.add(sql.toString() + "\r\n");
//            }
//            try {
//                writeFile(file.getAbsolutePath(), list);
//            } catch (Exception e) {
//                e.printStackTrace();
//            }
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}
